package com.shujia.indicators.core;

import com.shujia.indicators.utils.JdbcUtil;
import com.shujia.indicators.utils.SparkUtils;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

public class ConversionEtl {
    public static void main(String[] args) throws SQLException {
        SparkSession sparkSession = SparkUtils.initSession();

        Dataset<Row> orderMember = sparkSession.sql("select distinct(member_id) from bigdata005_user_portraits_dwd.dwd_t_order where order_status = 2");

        Dataset<Row> orderAgainMember = sparkSession.sql("select t.member_id as member_id" +
                " from (select count(order_id) as orderCount,member_id from bigdata005_user_portraits_dwd.dwd_t_order where order_status group by member_id ) as t where t.orderCount>=2");
        Dataset<Row> chargeMember = sparkSession.sql(
                "select distinct(member_id) as member_id from bigdata005_user_portraits_dim.dim_t_coupon_member " +
                        "where coupon_id !=1 and coupon_channel=1 "
        );
        Dataset<Row> chargeAndOrder = chargeMember.join(
                orderAgainMember,
                orderAgainMember.col("member_id").equalTo(chargeMember.col("member_id")),
                "inner"
        );
        long orderCount = orderMember.count();
        long orderAgainCount = orderAgainMember.count();
        long chargeCount = chargeAndOrder.count();
        Connection connection = null;
        PreparedStatement ps = null;
        try {
            connection = JdbcUtil.getConnection();
            ps = connection.prepareStatement("insert overwrite conversion values (?,?,?,?,?)");
            ps.setLong(1,1000L);
            ps.setLong(2,800L);
            ps.setLong(3,orderCount);
            ps.setLong(4,orderAgainCount);
            ps.setLong(5,chargeCount);
            ps.execute();
        }catch (Exception e) {
            e.printStackTrace();
        }finally {
            if (ps !=null){
            ps.close();
            }
            if (connection!=null){
                connection.close();
            }

        }



    }
}
